package com.ibm.cps.processors;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessageBuilder;
import com.ibm.cps.processors.vector.SchemaUtility;
import com.ibm.cps.sql.expression.ExpressionNode;
import com.ibm.cps.sql.expression.FieldEvaluator;
import com.ibm.cps.sql.expression.ValueExpression;
import com.ibm.cps.sql.expression.function.meta.FieldNameParam;
import com.ibm.cps.sql.expression.function.meta.IgnoredInFunctionCall;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.ErrorCode;
import com.ibm.util.exception.CPSException;

/**
 * Created by telekinesis on 7/2/15.
 * need one
 */
@Info(Name=MessageFields.VARIANCE_TYPE,Type=Info.ProcessorType.Vector)
public class VarianceProcessor extends SchemaInheritiveProcessor {
	@FieldNameParam
	private String inputField;
	@FieldNameParam @IgnoredInFunctionCall
	private String outputField;

	public VarianceProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
	}

	@Override
	public void dispose() {

	}

	@Override
	protected List<NewField> defineFieldsToBeAdded() {
		List<NewField> fields = new ArrayList<>();
		fields.add(new NewField(outputField, "double"));
		return fields;
	}


	@Override
	protected Collection<ParsedDataMessage> runProcessorLogic(ITimeSeriesSchema parentSchema,
			ITimeSeriesSchema outputSchema, ParsedDataMessage message) throws CPSException {
		Collection<ParsedDataMessage> messages = new ArrayList<>();
		ParsedDataMessageBuilder outputBuilder = ParsedDataMessageBuilder.createInhertiveBuilder(outputSchema, message, parentSchema);

		System.out.println("==============Variance input============");
		int index = parentSchema.getTagId(inputField);
		Double[] data = message.getArray(index);
		
		if (data == null || data.length == 0) {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, "the input was null");
		}
		
		double squareSum = 0;
		double sum = 0;
		for (int i = 0; i < data.length; i++) {
			double value = (double) data[i];
			sum += value;
			squareSum += value * value;
		}
		
		// variance = E(x^2) - (E(x))^2
		double avg = sum / (data.length);
		double variance = squareSum / (data.length) - avg * avg;

		outputBuilder.append(outputField, variance);
		System.out.println("==============Variance output============");
		System.out.println(Arrays.toString(this.getTimeSeriesSchema().getTagNames()));
		System.out.println(variance);
		
		messages.add(outputBuilder.toParsedDataMessage());
		return messages;
	}

	@Override
	public void initialize() throws CPSException {
		
	}
}
